package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object Demo1Sink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    studentDS.addSink(new MySinkFuntion2)

    env.execute()
  }

}

/**
  * 将数据发送出去的函数，每一条数据都会执行一次
  *
  * 都会建立连接
  */
class MySinkFuntion extends SinkFunction[String] {
  /**
    * invoke 将数据发送出去的函数，每一条数据都会执行一次
    * @param value   ds 的一行数据
    * @param context  上下文对象
    */
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    println(value)
  }
}

class MySinkFuntion2 extends RichSinkFunction[String] {

  /**
    * open 函数在invoke函数之前执行，每一个分区只执行一次
    *  arameters flink配置文件对象
    */
  var con: Connection = _
  override def open(parameters: Configuration): Unit = {
    println("正在建立数据库连接")
    //1、加载驱动


    Class.forName("com.mysql.jdbc.Driver")

    //22、建立链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/tour?useUnicode=true&characterEncoding=utf-8", "root", "123456")

  }

  //在invoke函数之后执行，每一个分区中只执行一次
  override def close(): Unit = {
    //关闭链接
    con.close()
  }
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {

    //将数据保存到mysql中
    val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

    val split: Array[String] = value.split(",")

    stat.setString(1, split(0))
    stat.setString(2, split(1))
    stat.setInt(3, split(2).toInt)
    stat.setString(4, split(3))
    stat.setString(5, split(4))

    //执行插入
    stat.execute()
  }
}